package com.kyrie.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.*;
import java.util.regex.Pattern;

/***
 * 描述: kafka消费者API源码学习
 * 源码：KafkaConsumer
 *
 * @author wuxiang
 * @date 2020-04-25 15:33
 */
public class KafkaConsumerAPI {

    public static void main(String[] args) {
        /**
         * 1.创建消费者
         */
        Map<String,Object> configs = new HashMap<>();
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"http://127.0.0.1:9092,http://127.0.0.1:9093,http://127.0.0.1:9094");
        configs.put(ConsumerConfig.GROUP_ID_CONFIG,"local-consumer-group");
        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        KafkaConsumer kafkaConsumer = new KafkaConsumer(configs);

        /*Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"http://127.0.0.1:9092,http://127.0.0.1:9093,http://127.0.0.1:9094");
        props.put(ConsumerConfig.GROUP_ID_CONFIG,"local-consumer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        KafkaConsumer kafkaConsumer = new KafkaConsumer(props);*/

        /**
         * 2.订阅主题topic: 可以通过集合的形式订阅多个主题，也可以通过正则的形式订阅特定模式的主题
         * ConsumerRebalanceListener 为再均衡监听器
         *
         *     # subscribe(Collection<String> topics, ConsumerRebalanceListener listener)
         *
         *     # subscribe(Collection<String> topics)
         *
         *     # subscribe(Pattern pattern, ConsumerRebalanceListener listener)
         *
         *     # subscribe(Pattern pattern)
         */

        // 集合
        /*kafkaConsumer.subscribe(Arrays.asList("topic-1","topic-2"));*/
        // 正则
        kafkaConsumer.subscribe(Pattern.compile("topic-*"));











    }


}
